package com.sduept.component.message.core;

import com.sduept.bigdata.fault.entity.FaultRecord;
import com.sduept.bigdata.fault.manager.FaultSourcePersistManager;
import com.sduept.core.dao.SuperDAO;

import lombok.extern.slf4j.Slf4j;
import net.sf.json.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Service
@Slf4j
public class MsgCallBackService {

	@Autowired
	private SuperDAO dao;
	@Autowired
	private FaultSourcePersistManager persistManager;
	
	private ExecutorService executor=Executors.newFixedThreadPool(5);

	@KafkaListener(topics = {"sf_wave_file_reget_callback"})
	public void onMessage(ConsumerRecord<?, ?> record) {
		Optional<?> kafkaMessage = Optional.ofNullable(record.value());
		if (kafkaMessage.isPresent()) {
				JSONObject jo = JSONObject.fromObject(kafkaMessage.get());
				String msgContent = jo.getString("content");
				JSONObject msgJson = JSONObject.fromObject(msgContent);
				String faultId=msgJson.getString("requestId");
				 log.info("接收到消息：sf_wave_file_reget_callback, msgContent："+msgContent+",faultId："+faultId);
				if(StringUtils.isNotEmpty(faultId)) {
					executor.submit(new Runnable() {
						@Override
						public void run() {
							final FaultRecord recordResult = dao.findById(faultId, FaultRecord.class);
							if(null==recordResult) {
								return;
							}
							persistManager.saveFaultSourceUploadInfo(recordResult);
						}
					});
				}
			}
			

	}
		
}
